%%%-------------------------------------------------------------------
%%% File    : rdf_sol_builder.erl
%%% Author  :  <vjache>
%%% Description : RDF solution set builder module.
%%%
%%% Created : 15 May 2009 by  <vjache>
%%%-------------------------------------------------------------------
-module(rdf_sol_builder).

-include("../include/rdf_spalg.hrl").
-include("../include/valg.hrl").
-include("../include/rdf_record.hrl"). 
-include("../include/pipe.hrl"). 

%% API
-export([
		 new_from_quad/1,
		 new_from_quad/2,
		 new_from_quad_by/2,
		 new_from_quad_by/3,
		 remap/2,
		 join_bf/3,
		 join_lk/3,
		 filter/2,
		 project/2,
		 index_for/1,
		 match/2
		 ]).
-define(DBG_DELAY,DBG_T=random:uniform(1000)+0,receive
					  after DBG_T ->
														ok
		end).
%% Helpers
%% -export([
%% 		 var_map_of/2,
%% 		 const_map_of/2,
%% 		 recount_var_map/1,
%% 		 sol_match/2,
%% 		 sol_match/3]).

%%====================================================================
%% API
%%====================================================================

%%--------------------------------------------------------------------
%% Function: new_from_quad(QuadPatt, RingMapByVar, NarrowByConst) -> solset()
%% Description: Creates a descriptor of a solution set based on quad 
%%				pattern. Narrows quad set based on specified constant 
%%				(represented at quad pattern) using effective index.
%%				Maps quads on ring according to specified variable (also 
%%				represented at quad pattern).
%%--------------------------------------------------------------------
new_from_quad_by(QuadPatt,
			  RingMapByVar,
			  NarrowByConst) when is_record(QuadPatt,?QUAD_TAG_SA) andalso
									  ?IS_VAR(RingMapByVar) andalso
									  ?IS_CONST(NarrowByConst) ->
	new_from_quad_by_(QuadPatt,
			   RingMapByVar,
			   {lookup,NarrowByConst}).
%%--------------------------------------------------------------------
%% Function: new_from_quad(QuadPatt, RingMapByVar) -> solset()
%% Description: Creates a descriptor of a solution set based on quad 
%%				pattern. Maps quads on ring according to specified 
%%				variable (represented at quad pattern).
%%--------------------------------------------------------------------
new_from_quad_by(QuadPatt,
			  RingMapByVar) when is_record(QuadPatt,?QUAD_TAG_SA) andalso
									  ?IS_VAR(RingMapByVar)  ->
	new_from_quad_by_(QuadPatt,
			   RingMapByVar,
			   all).
%%--------------------------------------------------------------------
%% Function: new_from_quad(QuadPatt,NarrowByConst) -> solset()
%% Description: Creates a descriptor of a solution set based on quad 
%%				pattern. Narrows quad set using lookup at index by 
%%				NarrowByConst const.
%%--------------------------------------------------------------------
new_from_quad(QuadPatt,NarrowByConst) when is_record(QuadPatt,?QUAD_TAG_SA),
										   ?IS_CONST(NarrowByConst) ->
	new_from_quad_(QuadPatt,{lookup,NarrowByConst}).
%%--------------------------------------------------------------------
%% Function: new_from_quad(QuadPatt,NarrowByConst) -> solset()
%% Description: Creates a descriptor of a solution set based on quad 
%%				pattern. Quad set matched over an entire dataset.
%%--------------------------------------------------------------------
new_from_quad(QuadPatt) when is_record(QuadPatt,?QUAD_TAG_SA) ->
	new_from_quad_(QuadPatt,all).
%%--------------------------------------------------------------------
%% Function: join_lk(SolSetBase,QuadPattList,JoinByVar) -> solset()
%% Description: Creates a descriptor of a solution set based on base 
%%				solution set and a list of quad	patterns. For each 
%%				base solution each quad	pattern produces quad stream 
%%				narrowed by value (bounded to JoinByVar variable in base 
%%				solution) using corresponding effecive index(lookup). 
%%				Then performed cartesian product to obtain a set of 
%%				solutions corresponding to value bounded at base sol, i.e.:
%%					[base_sol_i] x 
%%					lookup_quads_by(QP_1,get_binding(JoinByVar,base_sol_i)) x
%%					lookup_quads_by(QP_2,get_binding(JoinByVar,base_sol_i)) x
%%					... 											   x
%%					lookup_quads_by(QP_N,get_binding(JoinByVar,base_sol_i)).
%%--------------------------------------------------------------------
join_lk(SolSetBase,QuadPattList,JoinByVar) 
  when is_record(SolSetBase, solset) 
		   andalso ?IS_VAR(JoinByVar) ->
	BaseRemaped=case SolSetBase#solset.key_var of
					JoinByVar ->
						SolSetBase;
					_ ->
						remap(SolSetBase, JoinByVar)
				end,
	VectExprList=[
				  begin
					  VarRole=value_role_ex(JoinByVar, QuadPatt),
					  {?VECT_TAG,index_for(VarRole),
					   fun(Quad) ->
							   match(Quad, QuadPatt)%% -> SPOG ordered quad
					   end}
				  end ||
				  QuadPatt<-QuadPattList
				 ],
	VarMapAcc=var_map_of(QuadPattList,BaseRemaped#solset.var_map_acc, false),
	Var2PosList=var2pos_list(VarMapAcc),
	VarPositions=[Pos||{_,Pos}<-Var2PosList],
	Expr={?JOIN_LKP_TAG,
		  BaseRemaped#solset.expr,
		  VectExprList,
		  fun([SolBase|JoinedQuads]) ->
%% 				  ?DBG_DELAY,
				  SolAsList=tuple_to_list(SolBase)++[Node||{_,S,P,O,G}<-JoinedQuads,Node<-[S,P,O,G]],
		  		  Sol=list_to_tuple(SolAsList),
				  case sol_match(Sol, VarMapAcc,2) of
					  ?REJECT_REC ->
						  ?REJECT_REC;
					  _ ->
						  %%take only bindings of vars
						  list_to_tuple(['$sol',element(2,Sol)|[element(2+VPos,Sol)||VPos<-VarPositions]])
				  end
		  end
		 },
	#solset{
			expr=Expr,
			key_var=JoinByVar,
			var_map_acc=recount_var_map(VarMapAcc)
		   }.

%%--------------------------------------------------------------------
%% Function: remap(SolSet,ByVar) -> solset()
%% Description: Remaps on ring original solution set by specified variable. 
%%--------------------------------------------------------------------
remap(SolSet,ByVar) when is_record(SolSet, solset) andalso
							 ?IS_VAR(ByVar) ->
	case SolSet#solset.key_var of
		ByVar ->
			SolSet;
		_ ->
			VarMapAcc=SolSet#solset.var_map_acc,
			Pos=fetch_min_var_pos(ByVar, VarMapAcc),%%take most left position
			MapExpr={?MAP_TAG,
					 SolSet#solset.expr,
					 fun(SolRec) ->
							 erlang:setelement(2,SolRec,element(2+Pos,SolRec))
					 end},
			SolSet#solset{expr=MapExpr,key_var=ByVar}
	end.
%%--------------------------------------------------------------------
%% Function: join_bf(SolSetList,ByVar,JoinKind) -> solset()
%% Description: Brute force but massively parallel & distributed join 
%%				of a specified list solution sets by specified variable. 
%%--------------------------------------------------------------------
join_bf(SolSetList,ByVar,_JoinKind) when ?IS_VAR(ByVar)->
	Remapped=[remap(SolSet, ByVar)||SolSet<-SolSetList],
	VarMapTotAcc=lists:foldl(fun(SubSolSet,VarMap) ->
									 add_var_map(SubSolSet#solset.var_map_acc, VarMap)
							 end,{1,dict:new()},Remapped),
	Expr={?JOIN_TAG, 
	 [SolSet#solset.expr||SolSet<-Remapped], 
	 fun([Key|SolList]) ->
			 Sol=list_to_tuple(['$sol',Key|lists:foldr(fun(SubSol,Acc)->
								 ['$sol',Key|SubSolTail]=tuple_to_list(SubSol),
								 lists:foldr(fun(Binding,Bindings)->
													 [Binding|Bindings]
											 end, Acc, SubSolTail)
						 end,[],SolList)])
		  ,
			 sol_match(Sol,VarMapTotAcc,2)
	 end
	},
	#solset{
			expr=Expr,
			key_var=ByVar,
			var_map_acc=VarMapTotAcc
			}.
%%--------------------------------------------------------------------
%% Function: filter(SolSet,FilterExpr) -> solset()
%% Description: Filters specified solution set based on expression. 
%%--------------------------------------------------------------------
filter(SolSet,_FilterExpr) when is_record(SolSet, solset) ->
	ok.
%%--------------------------------------------------------------------
%% Function: project(SolSet,VarList) -> solset()
%% Description: Projects specified solution set based on specified 
%%				list of variables. 
%%--------------------------------------------------------------------
project(SolSet,VarList) when is_record(SolSet, solset) ->
	VarMapAcc={_,VarMap}=SolSet#solset.var_map_acc,
	VarPosList=[2+fetch_min_var_pos(Var,VarMapAcc)|| Var <- VarList],
	Expr={?FORMAT_TAG,
	 SolSet#solset.expr,
	 fun(Sol) ->
			 list_to_tuple(['$sol',_Key=element(2,Sol)|[element(Pos, Sol)||Pos<-VarPosList]])
	 end},
	VarMap1=dict:filter(fun(Var,_Pos) ->
						lists:member(Var, VarList)
				end, VarMap),
	SolSet#solset{
			expr=Expr,
			var_map_acc=recount_var_map({1,VarMap1})
			}.
fetch_min_var_pos(Var,{_,VarMap}=_VarMapAcc) when ?IS_VAR(Var)->
	lists:min(dict:fetch(Var, VarMap)).
%% distinct() ->
%% 	ok.
%% aggregate() ->
%% 	ok.


%%====================================================================
%% Internal functions
%%====================================================================

value_role_ex(Val, QuadPatt) when element(1,QuadPatt)==?QUAD_TAG_SA ->
	case value_role(Val, QuadPatt) of
		false ->
			exit({no_such_var,Val,QuadPatt});
		R -> R
	end.
value_role(Val,{?QUAD_TAG_SA,Val,_P,_O,_G}) ->
	s;
value_role(Val,{?QUAD_TAG_SA,_S,_P,_O,Val}) ->
	g;
value_role(Val,{?QUAD_TAG_SA,_S,_P,Val,_G}) ->
	o;
value_role(Val,{?QUAD_TAG_SA,_S,Val,_O,_G}) ->
	p;
value_role(_Val,_Q) ->
	false.

match(Q,QuadPatt) ->
	?SPOG=rdf_record:to_norm(Q),
	match(?SPOG,QuadPatt,5).
match(Quad,_QuadPatt,1) ->
	Quad;
match(Quad,QuadPatt,Pos) ->
	case element(Pos,QuadPatt) of
		Val when ?IS_VAR(Val) orelse Val=='_' ->
			match(Quad,QuadPatt,Pos-1);
		Val when ?IS_CONST(Val)->
			if 
				Val==element(Pos,Quad) ->
					match(Quad,QuadPatt,Pos-1);
				true ->
					?REJECT_REC
			end
	end.


%% sol_match(Sol,VarMapAcc) when element(1, Sol)=='$sol' ->
%% 	sol_match(Sol,VarMapAcc,2);
%% sol_match(Sol,VarMapAcc) ->
%% 	sol_match(Sol,VarMapAcc,0).
sol_match(Sol,{_Pos,VarMap}=VarMapAcc,ShiftPos) ->
	Var2PosList=var2pos_list(VarMapAcc),
	VarMapSize=dict:size(VarMap),
	Var2ValSet=sets:from_list([{Var,element(ShiftPos+Pos,Sol)}||{Var,Pos}<-Var2PosList]),
	case sets:size(Var2ValSet) of
		VarMapSize ->
			Sol;
		_ ->
%% 			io:format("DBG sol_match REJECT_REC ~p ~p ~p~n",[Sol,Var2PosList,sets:to_list(Var2ValSet)]),
%% 			receive
%% 				after 1000 ->
%% 					exit(test)
%% 			end,
			?REJECT_REC
	end.
%%--------------------------------------------------------------------
%% Function: index_for(ValueRole) -> vect()
%% Description: Chooses index (vector) for specified role of value in 
%%				quad.  ValueRole = atom() = s | p | o | g
%%--------------------------------------------------------------------
index_for(ValueRole) ->
	case svc_vms:get_vect(rdf_record:tag_by_role(ValueRole)) of
		{error,_}=ERR ->
			exit(ERR);
		Vect ->
			Vect
	end.

var_map_of(Arg,Distinct) ->
	var_map_of(Arg,{1,dict:new()},Distinct).
var_map_of({?QUAD_TAG_SA,S,P,O,G},Acc,Distinct) ->
	lists:foldl(
	  fun
		 (NodePatt,{Pos,VarMap}=AccIn) when ?IS_VAR(NodePatt) ->
			  case Distinct of
				  true ->
					  case dict:find(NodePatt, VarMap) of
						  {ok,_} ->
							  AccIn;
						  error ->
							  {Pos+1,dict:store(NodePatt, Pos, VarMap)}
					  end;
				  false ->
					  {Pos+1,dict:append(NodePatt, Pos, VarMap)}
			  end;
		 (NodePatt,{Pos,VarMap}) when ?IS_CONST(NodePatt) orelse NodePatt=='_' ->
			  {Pos+1,VarMap}
	  end, Acc, [S,P,O,G]);
var_map_of([]=_QuadPatternList,Acc,_Distinct) ->
	Acc;
var_map_of([QuadPattern|Tail]=_QuadPatternList,Acc,Distinct) ->
	Acc1=var_map_of(QuadPattern,Acc,Distinct),
	var_map_of(Tail,Acc1,Distinct).

%% const_map_of(Arg,Distinct) ->
%% 	{_,VarMap}=const_map_of(Arg,{1,dict:new()},Distinct),
%% 	VarMap.
%% const_map_of({?QUAD_TAG_SA,S,P,O,G},Acc,Distinct) ->
%% 	lists:foldl(
%% 	  fun
%% 		 (NodePatt,{Pos,VarMap}=AccIn) when ?IS_CONST(NodePatt) ->
%% 			  case Distinct of
%% 				  true ->
%% 					  case dict:find(NodePatt, VarMap) of
%% 						  {ok,_} ->
%% 							  AccIn;
%% 						  error ->
%% 							  {Pos+1,dict:store(NodePatt, Pos, VarMap)}
%% 					  end;
%% 				  false ->
%% 					  {Pos+1,dict:append(NodePatt, Pos, VarMap)}
%% 			  end;
%% 		 (NodePatt,{Pos,VarMap}) when ?IS_VAR(NodePatt) ->
%% 			  {Pos+1,VarMap}
%% 	  end, Acc, [S,P,O,G]);
%% const_map_of([]=_QuadPatternList,Acc,_Distinct) ->
%% 	Acc;
%% const_map_of([QuadPattern|Tail]=_QuadPatternList,Acc,Distinct) ->
%% 	Acc1=const_map_of(QuadPattern,Acc,Distinct),
%% 	const_map_of(Tail,Acc1,Distinct).

var2pos_list({_Pos,VarMap}=_Acc) ->
	lists:keysort(2, 
				  [{V,Pos}||
				   {V,PosList}<-dict:to_list(VarMap),Pos<-PosList]).
recount_var_map(Acc) ->
	add_var_map(Acc, 
				{1,dict:new()}).
add_var_map({_Pos,_VarMap}=Acc,{Pos1,VarMap1}=_AccTo) ->
	lists:foldl(fun({Node,_},{Pos,VarMapIn}) ->
						{Pos+1,dict:append(Node, Pos, VarMapIn)}
				end, 
				{Pos1,VarMap1}, 
				var2pos_list(Acc)).

new_from_quad_(QuadPatt,Query) when 
  is_record(QuadPatt,?QUAD_TAG_SA) 
	  andalso (Query==all orelse element(1,Query)==lookup) ->
	VarMapAcc=var_map_of(QuadPatt, false),
	VarPosList=[Pos||{_,Pos}<-var2pos_list(VarMapAcc)],
	RecFun=fun(Quad) -> % Record Fun & Filter in one.
							  case match(Quad,QuadPatt) of
								  ?REJECT_REC ->
									  ?REJECT_REC;
								  Spog -> %SPOG ordered quad
									  case sol_match(Spog, VarMapAcc,1) of
										  ?REJECT_REC ->
											  ?REJECT_REC;
										  _ ->
											  list_to_tuple(['$sol','$NULL'|
																 [element(Pos+1,Spog)||Pos<-VarPosList]])
									  end
							  end
					  end,
	VectExpr=case Query of
				 {lookup,NarrowByConst} ->
					 NarrowByRole=value_role_ex(NarrowByConst,QuadPatt),
					 {?VECT_TAG,
					  index_for(NarrowByRole),
					  {lookup,NarrowByConst},
					  RecFun};
				 all ->
					 {?VECT_TAG,
					  index_for(s),
					  RecFun}
			 end,
	#solset{expr=VectExpr,key_var='_',var_map_acc=recount_var_map(VarMapAcc)}.

new_from_quad_by_(QuadPatt,
			   RingMapByVar,
			   Query) when is_record(QuadPatt,?QUAD_TAG_SA) andalso
							   ?IS_VAR(RingMapByVar)  ->
	VectExpr=case Query of
				 {lookup,NarrowByConst} ->
					 NarrowByRole=value_role_ex(NarrowByConst,QuadPatt),
					 {?VECT_TAG,
					  index_for(NarrowByRole),
					  {lookup,NarrowByConst},
					  fun(Quad) -> % Record Fun & Filter in one.
							  match(Quad,QuadPatt)%-> SPOG ordered quad
					  end};
				 all ->
					 {?VECT_TAG,
					  index_for(s),
					  fun(Quad) -> % Record Fun & Filter in one.
							  match(Quad,QuadPatt)%-> SPOG ordered quad
					  end}
			 end,
	RingMapOrderTag= rdf_record:tag_by_role(value_role_ex(RingMapByVar,QuadPatt)),
	MapExpr	={?MAP_TAG,
			  VectExpr,
			  fun(Quad) ->
					  rdf_record:as(RingMapOrderTag, Quad)% reorder
			  end},
	VarMapAcc=var_map_of(QuadPatt, false),
	VarPosList=[Pos||{_,Pos}<-var2pos_list(VarMapAcc)],
	FormExpr={?FORMAT_TAG,
			  MapExpr,
			  fun(Quad) -> %% {'$sol',Key,Val1,...,ValN}
					  Spog=rdf_record:to_norm(Quad),
					  case sol_match(Spog, VarMapAcc,1) of
						  ?REJECT_REC ->
							  ?REJECT_REC;
						  _ ->
							  list_to_tuple(['$sol',element(2,Quad)|
												 [element(Pos+1,Spog)||Pos<-VarPosList]])
					  end
			  end},
	#solset{expr=FormExpr,key_var=RingMapByVar,var_map_acc=recount_var_map(VarMapAcc)}.

